package com.kkb.hbase2hdfs;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 需求（上）：读取上节创建的表myUser内容(f1,f2 列族)
 * k1  ImmutableBytesWritable   --> Rowkey
 * v1  Result
 * k2  keyout    ImmutableBytesWritable
 * v2  valueOut  Put
 */
public class HBaseReadMapper extends TableMapper<ImmutableBytesWritable, Put> {

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        byte[] rowKeyBytes = key.get();
        Put put = new Put(rowKeyBytes);
        //Return the array of Cells backing this Result instance.
        Cell[] cells = value.rawCells();  //上节使用的是listCells()
        for (Cell cell: cells){
            //获取列族名，Byte类型转String类型
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            //首先判断是 f1 f2 列族 if("f1".equals(family))
            if(StringUtils.containsAny(family,"f1","f2")){
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (StringUtils.containsAny(qualifier, "address","phone","name","age")) {
                    put.add(cell);
                }
            }

        }
        if(!put.isEmpty()){
            context.write(key,put);
        }
    }
}
